package com.lhf.aicloud.controller;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.lhf.aicloud.entity.Answer;
import com.lhf.aicloud.utils.AppVariable;
import com.lhf.aicloud.utils.idempotent.Idempotent;
import com.xxl.job.core.handler.annotation.XxlJob;
import jakarta.annotation.Resource;
import lombok.SneakyThrows;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.concurrent.TimeUnit;

/**
 * @author: lhf
 * @date: Created in 2024/08/15
 */
@RestController
@RequestMapping("/test")
public class TestController {

    @Resource
    private KafkaTemplate kafkaTemplate;

    private static final String TOPIC = "ai-cloud-test";

    @Resource
    private RedissonClient redissonClient;

    @RequestMapping("/test")
    public String test() throws InterruptedException {
        return "ok";
    }

    @RequestMapping("/testLock")
    public String testLock(Long uid) {
        String lockKey = AppVariable.getModelLockKey(uid, 1, 1);
        try {
            boolean isLock = redissonClient.getLock(lockKey).tryLock(2, TimeUnit.SECONDS);
            if (!isLock) {
                return "获取分布式锁失败";
            }
            System.out.println("获取锁成功");
//            Thread.sleep(3000);
            return "ok";
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        } finally {
            // 释放锁
            redissonClient.getLock(lockKey).unlock();
        }
    }

    @RequestMapping("/send")
    @Idempotent
    public String send(String msg) {
        kafkaTemplate.send(TOPIC, msg);
        return "ok";
    }

    @Idempotent
    @RequestMapping("/idempotentTest")
    public String idempotentTest(Integer id) {
        return "data: " + id;
    }

    @KafkaListener(topics = TOPIC)
    public void receive(String msg, Acknowledgment acknowledgment) {
        System.out.println("receive: " + msg);
        // 手动确认应答
        acknowledgment.acknowledge();
    }

    @XxlJob("testJob")
    public void testJob() {
        System.out.println("执行了定时任务");
    }

    @Resource
    private RedisTemplate<String, Object> redisTemplate;

    private static final String CANAL_TOPIC = "ai-cloud-canal-to-kafka";

    @Resource
    private ObjectMapper objectMapper;

    @SneakyThrows
    @KafkaListener(topics = CANAL_TOPIC)
    public void canal_listener(String msg, Acknowledgment acknowledgment) {
        System.out.println("receive: " + msg);

        HashMap<String, Object> map = objectMapper.readValue(msg, HashMap.class);
        if (!map.isEmpty() && "aicloud".equals(map.get("database").toString())
                && "answer".equals(map.get("table").toString())) {
            // 更新Redis缓存
            List<LinkedHashMap<String, Object>> list = (ArrayList<LinkedHashMap<String, Object>>) map.get("data");
            String cacheKey = "";
            for (LinkedHashMap<String, Object> answer : list) {
                cacheKey = AppVariable.getListCacheKey(
                        Long.parseLong(answer.get("uid").toString()),
                        Integer.parseInt(answer.get("model").toString()),
                        Integer.parseInt(answer.get("type").toString()));
                redisTemplate.opsForValue().set(cacheKey, null);
            }
        }
        // 手动确认应答
        acknowledgment.acknowledge();
    }
}
